package com.jvxb.demo.service.impl;

import com.jvxb.demo.service.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.KafkaAdminClient;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.KafkaFuture;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.List;
import java.util.Properties;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @author jvxb
 * @since 2023-01-15
 */
@Slf4j
@Service
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    KafkaTemplate kafkaTemplate;

    /**
     * 生产者端：指定往主题推送数据
     */
    @Override
    public void send(String topic, String message) {
        ListenableFuture listenableFuture = kafkaTemplate.send(topic, message);
        listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onSuccess(SendResult<String, String> result) {
                System.out.println(result);
                log.info("发送成功回调：{}", result.getProducerRecord().value());
            }

            @Override
            public void onFailure(Throwable ex) {
                log.info("发送失败回调", ex);
            }
        });
    }

    @Override
    public Set<String> listTopics() throws Exception {
        Properties pro = new Properties();
        List<String> bootstrapServers = (List<String>) kafkaTemplate.getProducerFactory().getConfigurationProperties().get("bootstrap.servers");
        pro.put("bootstrap.servers", bootstrapServers.stream().collect(Collectors.joining(",")));
        ListTopicsResult result = KafkaAdminClient.create(pro).listTopics();
        KafkaFuture<Set<String>> names = result.names();
        return names.get();
    }

    /**
     * 消费者端：消费监听的主题数据
     */
    @KafkaListener(topics = {"test_topic", "test_topic2"})
    public void handlerMsg(ConsumerRecord<String, String> consumerRecord) {
        System.out.println(consumerRecord);
        System.out.println("主题[test_topic]接收到消息：消息值：" + consumerRecord.value() + "， 消息偏移量：" + consumerRecord.offset());
    }
}
